package edu.npu.GraphIndex;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.Scanner;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

//负责读入标签选项并将标签附到节点上
public class TagGeneration {
	public static final String TAG_PATH = "hdfs://locker:9000/user/locker/tags";
	public static final String SEPARATOR = "\t";
	private static Random random = new Random(47);
	String[] tags = null;
	
	public TagGeneration(String[] tags) {
		this.tags = tags;
	}
	public TagGeneration(List<String> tags) {
		tags.toArray(this.tags);
	}
	public String next() {
		return tags[random.nextInt(tags.length)];
	}
	public static class TagGenerationMapper extends Mapper<Text, Text, Text, Text> {
		List<String> tags = new ArrayList<String>();
		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			Configuration conf = new Configuration();
			FileSystem fs = FileSystem.get(conf);
			Scanner scanner = new Scanner(fs.open(new Path(TAG_PATH)));
			while(scanner.hasNext()) {
				tags.add(scanner.nextLine());
			}
		}

		@Override
		protected void map(Text key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] splits = line.split(SEPARATOR);
			TagGeneration tg = new TagGeneration(tags);
			context.write(new Text(splits[0]), new Text(tg.next()));
		}
	}
	public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
		Configuration conf = new Configuration();
		Job job = new Job(conf);
		job.setJarByClass(TagGeneration.class);
		job.setMapperClass(TagGeneration.TagGenerationMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path("id_tag"));
		job.waitForCompletion(true);
	}
}

